package fr.inserm.um915.bdibd.model;

import java.awt.Point;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.logging.Logger;
import java.util.regex.Pattern;


import com.sleepycat.bind.tuple.IntegerBinding;
import com.sleepycat.bind.tuple.StringBinding;
import com.sleepycat.bind.tuple.TupleInput;
import com.sleepycat.bind.tuple.TupleOutput;
import com.sleepycat.je.Cursor;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;
import com.sleepycat.je.JoinCursor;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;
import com.sleepycat.je.SecondaryConfig;
import com.sleepycat.je.SecondaryCursor;
import com.sleepycat.je.SecondaryDatabase;
import com.sleepycat.je.SecondaryKeyCreator;
import com.sleepycat.je.Transaction;

/**
 * Model
 *
 */
public class Model
	{
	private static final Logger LOG=Logger.getLogger("fr.inserm.um915.bdibd");
	private static final String DB_ID2INDIVIDUAL_NAME="id2individual";
	private static final String DB_FAM2INDIVIDUAL_NAME="fam2individual";
	private static final String DB_PAIRID2INDIVIDUALIDS="pairIdIndividualId";
	private static final String DB_INDI1OFPAIR="indi1ofPair";
	private static final String DB_INDI2OFPAIR="indi2ofPair";
	private static final String DB_MARKER2LINKAGE="marker2linkage";
	private static final String DB_POS2MARKER="pos2marker";
	private static final String DB_NAME2MARKER="rs2marker";
	private static final String DB_INDEX2RS="index2rs";
	private static final String DB_PAIRID2THEO="pairidtotheo";
	private static final String DB_META="meta";
	private Environment environment=null;
	private Database id2individual=null;
	private SecondaryDatabase family2individual=null;
	private Database pairIdIndividualId=null;
	private SecondaryDatabase indi1ofpair=null;
	private SecondaryDatabase indi2ofpair=null;
	private Database marker2linkage=null;
	private Database pos2marker=null;
	private SecondaryDatabase name2marker=null;
	private SecondaryDatabase index2rs=null;
	private Database pairid2theo=null;
	private Database metadata=null;
	
	private int count_markers=0;
	private int count_pairs=0;
	private int count_individuals=0;
	private boolean testing_flag=false;
	
	/** Model */
	public Model()
		{
			
		}
	
	Database getPosToMarkerDatabase()
		{
		return pos2marker;
		}
	
	/**
	 * open the model
	 * @param dir berkeleyDB home
	 * @param readOnly is the database read-only
	 * @throws IOException
	 */
	public void open(File dir,boolean readOnly) throws IOException
		{
		Transaction txn=null;
		close();
		LOG.info("opening "+dir);
		EnvironmentConfig envCfg=new EnvironmentConfig();
		envCfg.setAllowCreate(!readOnly);
		envCfg.setReadOnly(readOnly);
		if(!readOnly) envCfg.setCachePercent(80);
		envCfg.setConfigParam(EnvironmentConfig.LOG_FILE_MAX,"300000000");
		/** see http://www.oracle.com/technetwork/database/berkeleydb/je-faq-096044.html#CanBerkeleyDBJavaEditionuseaNFSSANorotherremotesharednetworkfilesystemforanenvironment */
		envCfg.setConfigParam(EnvironmentConfig.LOG_USE_ODSYNC, "true");
		//if(!readOnly) envCfg.setLocking(false);
		this.environment=new Environment(dir, envCfg);
		
		if(!readOnly)
			{
			for(String db:this.environment.getDatabaseNames())
				{
				LOG.info("drop "+db);
				this.environment.truncateDatabase(txn, db, false);
				}
			
			}
		
		
		
		DatabaseConfig cfg=new DatabaseConfig();
		cfg.setAllowCreate(!readOnly);
		cfg.setReadOnly(readOnly);
		cfg.setBtreeComparator(SortedIntegerComparator.class);
		this.id2individual=this.environment.openDatabase(txn, DB_ID2INDIVIDUAL_NAME, cfg);
		
		SecondaryConfig cfg2=new SecondaryConfig();
		cfg2.setAllowCreate(!readOnly);
		cfg2.setReadOnly(readOnly);
		cfg2.setSortedDuplicates(true);
		cfg2.setKeyCreator(new SecondaryKeyCreator()
			{
			@Override
			public boolean createSecondaryKey(
					SecondaryDatabase secondary,
                    DatabaseEntry key,
                    DatabaseEntry data,
                    DatabaseEntry result)
				{
				Individual indi= Individual.BINDING.entryToObject(data);
				StringBinding.stringToEntry(indi.getFamily(), result);
				return true;
				}
			});
		cfg2.setBtreeComparator(IStringComparator.class);
		this.family2individual=this.environment.openSecondaryDatabase(txn, DB_FAM2INDIVIDUAL_NAME, id2individual,cfg2);
		
		
		
		cfg=new DatabaseConfig();
		cfg.setAllowCreate(!readOnly);
		cfg.setReadOnly(readOnly);
		cfg.setBtreeComparator(SortedIntegerComparator.class);
		this.pairIdIndividualId=this.environment.openDatabase(txn, DB_PAIRID2INDIVIDUALIDS, cfg);
		
		
		cfg2=new SecondaryConfig();
		cfg2.setAllowCreate(!readOnly);
		cfg2.setReadOnly(readOnly);
		cfg2.setSortedDuplicates(true);
		cfg2.setKeyCreator(new SecondaryKeyCreator()
			{
			@Override
			public boolean createSecondaryKey(
					SecondaryDatabase secondary,
                    DatabaseEntry key,
                    DatabaseEntry data,
                    DatabaseEntry result)
				{
				TupleInput in=new TupleInput(data.getData());
				int n=in.readInt();
				IOUtils.safeClose(in);
				IntegerBinding.intToEntry(n, result);
				return true;
				}
			});
		cfg2.setBtreeComparator(SortedIntegerComparator.class);
		this.indi1ofpair=this.environment.openSecondaryDatabase(txn, DB_INDI1OFPAIR, pairIdIndividualId,cfg2);
		
		cfg2=new SecondaryConfig();
		cfg2.setAllowCreate(!readOnly);
		cfg2.setReadOnly(readOnly);
		cfg2.setSortedDuplicates(true);
		cfg2.setKeyCreator(new SecondaryKeyCreator()
			{
			@Override
			public boolean createSecondaryKey(
					SecondaryDatabase secondary,
                    DatabaseEntry key,
                    DatabaseEntry data,
                    DatabaseEntry result)
				{
				TupleInput in=new TupleInput(data.getData());
				in.readInt();//skip first
				int n=in.readInt();
				IOUtils.safeClose(in);
				IntegerBinding.intToEntry(n, result);
				return true;
				}
			});
		cfg2.setBtreeComparator(SortedIntegerComparator.class);
		this.indi2ofpair=this.environment.openSecondaryDatabase(txn, DB_INDI2OFPAIR, pairIdIndividualId,cfg2);
		
		
		cfg=new DatabaseConfig();
		cfg.setAllowCreate(!readOnly);
		cfg.setReadOnly(readOnly);
		//cfg.setDeferredWrite(!readOnly);
		this.marker2linkage = this.environment.openDatabase(txn, DB_MARKER2LINKAGE, cfg);
		
		cfg=new DatabaseConfig();
		cfg.setAllowCreate(!readOnly);
		cfg.setReadOnly(readOnly);
		cfg.setBtreeComparator(ChromPositionComparator.class);
		//cfg.setDeferredWrite(!readOnly);
		this.pos2marker = this.environment.openDatabase(txn, DB_POS2MARKER, cfg);
		
		cfg2=new SecondaryConfig();
		cfg2.setAllowCreate(!readOnly);
		cfg2.setReadOnly(readOnly);
		cfg2.setKeyCreator(new SecondaryKeyCreator()
			{
			@Override
			public boolean createSecondaryKey(
					SecondaryDatabase secondary,
                    DatabaseEntry key,
                    DatabaseEntry data,
                    DatabaseEntry result)
				{
				Marker m=Marker.BINDING.entryToObject(data);
				StringBinding.stringToEntry(m.getName(), result);
				return true;
				}
			});
		//cfg2.setDeferredWrite(!readOnly);
		this.name2marker=this.environment.openSecondaryDatabase(txn, DB_NAME2MARKER, pos2marker,cfg2);
		
		cfg2=new SecondaryConfig();
		cfg2.setAllowCreate(!readOnly);
		cfg2.setReadOnly(readOnly);
		cfg2.setKeyCreator(new SecondaryKeyCreator()
			{
			@Override
			public boolean createSecondaryKey(
					SecondaryDatabase secondary,
                    DatabaseEntry key,
                    DatabaseEntry data,
                    DatabaseEntry result)
				{
				Marker m=Marker.BINDING.entryToObject(data);
				IntegerBinding.intToEntry(m.index, result);
				return true;
				}
			});
		cfg2.setBtreeComparator(SortedIntegerComparator.class);
		//cfg2.setDeferredWrite(!readOnly);
		this.index2rs = this.environment.openSecondaryDatabase(txn, DB_INDEX2RS, pos2marker,cfg2);
		
		
		cfg=new DatabaseConfig();
		cfg.setAllowCreate(!readOnly);
		cfg.setReadOnly(readOnly);
		cfg.setBtreeComparator(SortedIntegerComparator.class);
		this.pairid2theo=this.environment.openDatabase(txn, DB_PAIRID2THEO, cfg);
		
		cfg=new DatabaseConfig();
		cfg.setAllowCreate(!readOnly);
		cfg.setReadOnly(readOnly);
		cfg.setBtreeComparator(IStringComparator.class);
		this.metadata=this.environment.openDatabase(txn, DB_META, cfg);
		
		
		if(readOnly)
			{
			this.count_markers=_readDBCount(txn, this.index2rs);
			this.count_pairs=_readDBCount(txn, this.pairIdIndividualId);
			this.count_individuals=_readDBCount(txn, this.id2individual);
			}
		else
			{
			SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
			setMeta(txn, "creation.date", sdf.format(Calendar.getInstance().getTime()));
			setMeta(txn, "user.name",System.getProperty("user.name"));
			setMeta(txn, "user.dir",System.getProperty("user.dir"));
			}
		}
	
	
	
	/** close env */
	public void close()
		{
		this.count_markers=0;
		this.count_individuals=0;
		this.count_pairs=0;
		
		if(index2rs!=null)
			{
			try{ this.index2rs.close(); }
			catch(Throwable err) { }
			this.index2rs=null;
			}
		
		if(name2marker!=null)
			{
			try{ this.name2marker.close(); }
			catch(Throwable err) { }
			this.name2marker=null;
			}
		
		if(pos2marker!=null)
			{
			try{ this.pos2marker.close(); }
			catch(Throwable err) { }
			this.pos2marker=null;
			}
		
		if(this.family2individual!=null)
			{
			try{ this.family2individual.close(); }
			catch(Throwable err) { }
			this.family2individual=null;
			}
		
		if(this.id2individual!=null)
			{
			try{ this.id2individual.close(); }
			catch(Throwable err) { }
			this.id2individual=null;
			}
		
		
		if(this.indi1ofpair!=null)
			{
			try{ this.indi1ofpair.close(); }
			catch(Throwable err) { }
			this.indi1ofpair=null;
			}
		if(this.indi2ofpair!=null)
			{
			try{ this.indi2ofpair.close(); }
			catch(Throwable err) { }
			this.indi2ofpair=null;
			}
		
		if(this.pairIdIndividualId!=null)
			{
			try{ this.pairIdIndividualId.close(); }
			catch(Throwable err) { }
			this.pairIdIndividualId=null;
			}
		
		
		if(this.marker2linkage!=null)
			{
			try{ this.marker2linkage.close(); }
			catch(Throwable err) { }
			this.marker2linkage=null;
			}
		
		if(this.pairid2theo!=null)
			{
			try { this.pairid2theo.close();}
			catch(Throwable err) { }
			this.pairid2theo=null;
			}
		
		
		if(this.metadata!=null)
			{
			try { this.metadata.close();}
			catch(Throwable err) { }
			this.metadata=null;
			}
		
		if(this.environment!=null)
			{
			if(!this.environment.getConfig().getReadOnly())
				{
				this.environment.sync();
				this.environment.cleanLog();
				this.environment.compress();
				}
			this.environment.close();
			this.environment=null;
			}
		}
	
	public int getIndividualCount()
		{
		return this.count_individuals;
		}
	
	public int getMarkerCount()
		{
		return this.count_markers;
		}
	
	public int getPairCount()
		{
		return this.count_pairs;
		}
	
	private int _readDBCount(Transaction txn,Database db)
		{
		Cursor c=null;
		try {
			DatabaseEntry key1=new DatabaseEntry();
			DatabaseEntry data1=new DatabaseEntry();
			c= db.openCursor(txn, null);
			if(c.getLast(key1, data1, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
				{
				throw new RuntimeException("boum");
				}
			int max_index=IntegerBinding.entryToInt(key1);
			return max_index+1;
			}
		finally
			{
			if(c!=null) c.close();
			}
		}
	
	/** @return all individuals that belong to a given family */
	public List<Individual> getIndividualsByFamily(Transaction txn,String famname)
		{
		SecondaryCursor c=null;
		List<Individual> indis=new ArrayList<Individual>();
		try {
			DatabaseEntry key1=new DatabaseEntry();
			StringBinding.stringToEntry(famname, key1);
			DatabaseEntry data1=new DatabaseEntry();
			c=this.family2individual.openCursor(txn, null);
			boolean first=true;
			while(
				(first?
					c.getSearchKey(key1, data1, LockMode.DEFAULT):
					c.getNextDup(key1, data1, LockMode.DEFAULT))==OperationStatus.SUCCESS)
				{
				first=false;
				indis.add(Individual.BINDING.entryToObject(data1));
				}
			return indis;
			} 
		catch (Exception e)
			{
			throw new RuntimeException(e);
			}
		finally
			{
			if(c!=null) c.close();
			}
		}
	
	/** @return meta data for the given key or null */
	public String getMeta(Transaction txn,String k)
		{
		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry data=new DatabaseEntry();
		StringBinding.stringToEntry(k, key);
		if(metadata.get(txn, key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
			{
			return null;
			}
		return StringBinding.entryToString(data);
		}
	
	/** @return all metadata as an object 'Properties' */
	public Map<String, String> getMeta(Transaction txn)
		{
		TreeMap<String, String> p=new TreeMap<String, String>();
		Cursor c=null;
		try {
			DatabaseEntry key1=new DatabaseEntry();
			DatabaseEntry data1=new DatabaseEntry();
			c=this.metadata.openCursor(txn, null);
			while(c.getNext(key1, data1, LockMode.DEFAULT)==OperationStatus.SUCCESS)
				{
				p.put(
					StringBinding.entryToString(key1),
					StringBinding.entryToString(data1)
					);
				}
			return p;
			} 
		catch (Exception e)
			{
			throw new RuntimeException(e);
			}
		finally
			{
			if(c!=null) c.close();
			}
		}
	
	/** set a meta data */
	private void setMeta(Transaction txn,String k,String v)
		{
		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry data=new DatabaseEntry();
		StringBinding.stringToEntry(k, key);
		StringBinding.stringToEntry(v==null?"null":v,data);
		if(metadata.put(txn, key, data)!=OperationStatus.SUCCESS)
			{
			throw new RuntimeException("Cannot insert meta "+k+" "+v);
			}
		}
	
	/**
	 * answers a marker by its name
	 * @return a marker or null */
	public Marker getMarkerByName(
			Transaction txn,String rsName)
		{
		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry data=new DatabaseEntry();
		StringBinding.stringToEntry(rsName, key);
		if(name2marker.get(txn, key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
			{
			return null;
			}
		return Marker.BINDING.entryToObject(data);
		}
	
	/**
	 * answers a marker by its genomic position
	 * @return a marker or null */
	public Marker getMarkerByPosition(
			Transaction txn,ChromPosition pos)
		{
		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry data=new DatabaseEntry();
		ChromPosition.BINDING.objectToEntry(pos, key);
		if(pos2marker.get(txn, key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
			{
			return null;
			}
		return Marker.BINDING.entryToObject(data);
		}
	
	/**
	 * answers a marker by its id
	 * @return a marker or null */
	public Marker getMarkerByIndex(
			Transaction txn,int index)
		{
		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry data=new DatabaseEntry();
		IntegerBinding.intToEntry(index, key);
		if(index2rs.get(txn, key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
			{
			return null;
			}
		return Marker.BINDING.entryToObject(data);
		}
	
	/**
	 *  returns the lower bound index of the marker having a position greater or equals
	 *  to the given position
	 * @param txn
	 * @param pos
	 * @return the marker index or count_markers
	 */
	public int getMarkerIndexLowerBound(Transaction txn,ChromPosition pos)
		{
		Cursor c=null;
		try
			{
			DatabaseEntry key=new DatabaseEntry();
			DatabaseEntry data=new DatabaseEntry();
			ChromPosition.BINDING.objectToEntry(pos,key);
			c=pos2marker.openCursor(txn, null);
			if(c.getSearchKeyRange(key, data, LockMode.DEFAULT)==OperationStatus.SUCCESS)
				{
				return Marker.BINDING.entryToObject(data).index;
				}
			return this.count_markers;
			}
		finally
			{
			if(c!=null) c.close();
			}
		}
	
	
	public IBDStatus getTheoriticalIBDStatusByPairId(
			Transaction txn,int index)
		{
		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry data=new DatabaseEntry();
		IntegerBinding.intToEntry(index, key);
		if(pairid2theo.get(txn, key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
			{
			return null;
			}
		return IBDStatus.BINDING.entryToObject(data);
		}
	/**
	 * answers a pair of individual from a pair_id
	 * @param txn
	 * @param pairid
	 * @return a pair of individual or null
	 */
	public PairOfIndividual getPairOfIndividualByPairId(
			Transaction txn,int pairid)
		{
		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry data=new DatabaseEntry();
		IntegerBinding.intToEntry(pairid, key);
		if(pairIdIndividualId.get(txn, key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
			{
			return null;
			}
		TupleInput input=new TupleInput(data.getData(),data.getOffset(),data.getSize());
		int i1=input.readInt();
		int i2=input.readInt();
		IOUtils.safeClose(input);
		return new PairOfIndividual(
				getIndividualByIndex(txn, i1),
				getIndividualByIndex(txn, i2)
				);
		}
	
	
	/**
	 * answers the IBD-status for the given rs## and the given pair_id
	 * @param txn
	 * @param rsName
	 * @param pair_id
	 * @return the IBD status or null
	 */
	public IBDStatus getIbdStatus(
			Transaction txn,String rsName,
			int pair_id
			)
		{
		LinkageRow L=getIBDByMarker(txn,rsName);
		if(L==null || L.size()<=pair_id) return new IBDStatus();
		return L.get(pair_id);
		}
	
	
	/**
	 * answers an individual from its id
	 * @param txn
	 * @param index
	 * @return the individual or null
	 */
	public Individual getIndividualByIndex(Transaction txn,int index)
		{
		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry data=new DatabaseEntry();
		IntegerBinding.intToEntry(index, key);
		if(id2individual.get(txn, key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
			{
			return null;
			}
		return Individual.BINDING.entryToObject(data);
		}
	
	/**
	 * returns the full row of IBD status by its rsName or null
	 * @param txn
	 * @param rsName
	 * @return the array of IBD
	 */
	public LinkageRow getIBDByMarker(Transaction txn,String rsName)
		{
		try {
			DatabaseEntry key1=new DatabaseEntry();
			StringBinding.stringToEntry(rsName, key1);
			DatabaseEntry data1=new DatabaseEntry();
			if(this.marker2linkage.get(txn, key1, data1, LockMode.DEFAULT)!=OperationStatus.SUCCESS)
				{
				//LOG.info("no linkage data for "+rsName);
				return null;
				}
			return LinkageRow.BINDING.entryToObject(data1);
			} 
		catch (Exception e)
			{
			throw new RuntimeException(e);
			}
		finally
			{
		
			}
		}
	
	/**
	 * answers an individual from its family and its name
	 * @param txn
	 * @param family
	 * @param name
	 * @return the individual
	 */
	public Individual getIndividualbyName(Transaction txn,String family,String name)
		{
		SecondaryCursor c=null;
		
		try {
			DatabaseEntry key1=new DatabaseEntry();
			StringBinding.stringToEntry(family, key1);
			DatabaseEntry data1=new DatabaseEntry();
			c=this.family2individual.openCursor(txn, null);
			boolean first=true;
			while(
				(first?
					c.getSearchKey(key1, data1, LockMode.DEFAULT):
					c.getNextDup(key1, data1, LockMode.DEFAULT))==OperationStatus.SUCCESS)
				{
				first=false;
				Individual indi=Individual.BINDING.entryToObject(data1);
				if(!indi.getFamily().equalsIgnoreCase(family)) throw new RuntimeException();
				if(indi.name.equalsIgnoreCase(name)) return indi;
				}
			return null;
			} 
		catch (Exception e)
			{
			throw new RuntimeException(e);
			}
		finally
			{
			if(c!=null) c.close();
			}
		}
	/**
	 * find pair per individual
	 * @param txn
	 * @param i1
	 * @param i2
	 * @return pair-id or -1
	 */
	public int findPairIdByIndividuals(Transaction txn,Individual i1,Individual i2)
		{
		return findPairIdByIndividuals(txn,i1.id,i2.id);
		}
	
	public int findPairIdByIndividuals(Transaction txn,int indi1,int indi2)
		{
		if(indi1>indi2) return findPairIdByIndividuals(txn,indi2,indi1);
		SecondaryCursor c1 = null;
	    SecondaryCursor c2 = null;
	    JoinCursor joinCursor = null;
	    DatabaseEntry key = new DatabaseEntry();
        DatabaseEntry data = new DatabaseEntry();
        
	    try
	    	{
	    	c1 = indi1ofpair.openCursor(txn, null);
	        c2 = indi2ofpair.openCursor(txn, null);
	        SecondaryCursor[] cursors = {c1, c2};
	        IntegerBinding.intToEntry(indi1, key);
	        if(c1.getSearchKey(key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS) return -1;
	        IntegerBinding.intToEntry(indi2, key);
	        if(c2.getSearchKey(key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS) return -1;
	        joinCursor = pairIdIndividualId.join(cursors, null);
	        while (true)
	        	{
	        	if(joinCursor.getNext(key, data, LockMode.DEFAULT)!=OperationStatus.SUCCESS) break;
	        	return IntegerBinding.entryToInt(key);
	        	}
	    	}
	    catch(Exception err)
	    	{
	    	throw new RuntimeException(err);
	    	}
	    finally
	    	{
	    	if(c1!=null) c1.close();
	    	if(c2!=null) c2.close();
	    	if(joinCursor!=null) joinCursor.close();
	    	}
	     
		return -1;
		}
	
	
	private void doLayout(
			List<Individual> indis,
			File dotExecPath
			) throws IOException
		 {
		LOG.info("Layout");
		 File tmpout=null;
         PrintWriter fout=null;
         BufferedReader in=null;
         Map<String,Individual> name2indi=new HashMap<String,Individual>();
         Map<Integer,Individual> id2indi=new HashMap<Integer,Individual>();
         
         for(Individual indi:indis)
         	{
        	 name2indi.put(indi.getName(),indi);
        	 id2indi.put(indi.getId(),indi);
         	}
         
         try
	         {
			 tmpout=File.createTempFile("graphviz", ".dot");
			 fout=new PrintWriter(tmpout);
	         
			 
			 fout.write("digraph G{");
             for(Individual indi:indis)
                     {
            	 	fout.write("i"+indi.getId()+";\n");
                     
                     if(indi.hasFather())
                             {
                    	 	 Individual parent=name2indi.get(indi.getFatherId());
                             if(parent==null)  throw new RuntimeException(indi.toString());
                             fout.write("i"+indi.getId()+" -> i"+parent.getId()+" ;\n");
                             }
                     if(indi.hasMother())
                             {
                    	 	 Individual parent=name2indi.get(indi.getMotherId());
                    	 	 if(parent==null)  throw new RuntimeException(indi.toString());
                    	 	fout.write("i"+indi.getId()+" -> i"+parent.getId()+" ;\n");
                             }
                     }
             fout.write("}");
             

			 
	         fout.flush();
	         fout.close();
	         
	         List<String> args=new ArrayList<String>();
	         args.add(dotExecPath.toString());
	         
	         
	         args.add(tmpout.getAbsolutePath());
	         Process proc=Runtime.getRuntime().exec(args.toArray(new String[args.size()]));
	         in=new BufferedReader(new InputStreamReader(proc.getInputStream()));
	         
	         
	         final int scale=3;
             String line;
             while((line=in.readLine())!=null)
                     {
                     
                     line=line.trim();
                     
                     int n=line.indexOf('[');
                     int indexPos=line.indexOf("pos=\"");
                     if(n==-1) continue;
                     
                     if(line.contains("->") || indexPos==-1 || !line.contains("width") || !line.contains("height")) continue;
                     String id=line.substring(0,n).trim();
                     if(!id.startsWith("i")) continue;
                     id=id.substring(1);
                     Individual individual =id2indi.get(Integer.parseInt(id));
                     if(individual==null) continue;
                     int end=line.indexOf("\"",indexPos+5);
                     if(end==-1) continue;
                     int comma= line.indexOf(",",indexPos+5);
                     if(comma==-1) continue;
                     
                     int x= Integer.parseInt(line.substring(indexPos+5,comma));
                     int y= Integer.parseInt(line.substring(comma+1,end));
                     individual.pedPt=new Point((int)(x*scale),(int)(y*scale));
                     LOG.info("Individual "+individual+ " "+individual.pedPt);
                     }
	         
	         in.close();
	         }
         catch(Exception err)
         	{
        
        	IOUtils.safeClose(fout);
        	IOUtils.safeClose(in);
        	 if(tmpout!=null) tmpout.delete();

         	}
		}
	/**
	 * readPedigree
	 * @param txn
	 * @param ped
	 * @param graphvizExe
	 * @throws IOException
	 */
	public void readPedigree(Transaction txn,File ped,File graphvizExe) throws IOException
		{
		this.count_individuals=0;
		setMeta(txn,"pedigree.file",ped.getAbsolutePath());
		Pattern tab=Pattern.compile("[\t ]");
		BufferedReader in=IOUtils.open(ped);
		String line;
		List<Individual> individuals=new ArrayList<Individual>();
		Set<String> families= new HashSet<String>();
		while((line=in.readLine())!=null)
			{
			if(line.length()==0) continue;
			String tokens[]=tab.split(line);
			if(tokens.length!=6)
				{
				for(int i=0;i< tokens.length;++i) System.err.println("("+(i+1)+")"+tokens[i]);
				throw new IOException(line+" . expected 6 columns , got "+ tokens.length);
				}
			Individual indi=new Individual();
			
			indi.family=tokens[0];
			families.add(indi.family);//for graphviz dot
			indi.name=tokens[1];
			indi.fatherId=tokens[2];
			indi.motherId=tokens[3];
			indi.gender=Integer.parseInt(tokens[4]);
			indi.status=Integer.parseInt(tokens[5]);
			individuals.add(indi);
			}
		IOUtils.safeClose(in);
		if(individuals.isEmpty()) throw new RuntimeException("pedigree empty");
		
		
		LOG.info("sorting "+ individuals.size()+" individuals.");
		Collections.sort(individuals);
		setMeta(txn, "pedigree.size", String.valueOf(individuals.size()));
		
		for(Individual indi:individuals)
			{
			indi.id=count_individuals;
			count_individuals++;
			}
		
		
		if(graphvizExe!=null)
			{
			setMeta(txn, "graphviz.dot", graphvizExe.toString());
			for(String fam:families)
				{
				List<Individual> famindi=new ArrayList<Individual>();
				for(Individual indi:individuals)
					{
					if(!indi.getFamily().equals(fam)) continue;
					famindi.add(indi);
					}
				doLayout(famindi,graphvizExe);
				}
			}
		

		DatabaseEntry key=new DatabaseEntry();
		DatabaseEntry value=new DatabaseEntry();
		
		
		for(Individual indi:individuals)
			{
			IntegerBinding.intToEntry(indi.id, key);
			Individual.BINDING.objectToEntry(indi, value);
			if(this.id2individual.put(txn, key, value)!=OperationStatus.SUCCESS)
				{
				throw new RuntimeException("Cannot insert "+indi);
				}
			}
		
		}
	
	private void readDataFile(Transaction txn,File dataFile) throws IOException
		{
		Cursor cursor=null;
		try
			{
			long start=System.currentTimeMillis();
			LOG.info(dataFile.toString());
			Pattern tab=Pattern.compile("[\t ]");
			BufferedReader in=IOUtils.open(dataFile);
			cursor=this.marker2linkage.openCursor(txn, null);
			String line;
			Individual prev_indi1=null;
			Individual prev_indi2=null;
			int prev_pair_id=-1;
			int nLine=0;
			long nInserted=0;
			while((line=in.readLine())!=null)
				{
				++nLine;
				if(line.length()==0) continue;
				if(nLine>1000 && testing_flag) break;
				String tokens[]=tab.split(line);
				if(tokens.length!=9) throw new IOException(line+" "+tokens.length);
				Individual indi1=null;
				if(prev_indi1!=null && prev_indi1.family.equals(tokens[0]) && prev_indi1.name.equals(tokens[1]))
					{
					indi1=prev_indi1;
					}
				else
					{
					indi1=getIndividualbyName(txn, tokens[0],tokens[1]);
					if(indi1==null) LOG.info("no info about "+tokens[0]+":"+tokens[1]);
					prev_indi1=indi1;
					prev_pair_id=-1;
					}
				Individual indi2=null;
				if(prev_indi2!=null && prev_indi2.family.equals(tokens[2]) && prev_indi2.name.equals(tokens[3]))
					{
					indi2=prev_indi2;
					}
				else
					{
					indi2=getIndividualbyName(txn, tokens[2],tokens[3]);
					if(indi2==null) LOG.info("no info about "+tokens[2]+":"+tokens[3]);
					prev_indi2=indi2;
					prev_pair_id=-1;
					}
				int pair_id=prev_pair_id;
				if(pair_id<0)
					{
					pair_id=findPairIdByIndividuals(txn,indi1,indi2);
					if(pair_id<0)
						{
						pair_id=this.count_pairs;
						LOG.info(line+" "+pair_id);
						DatabaseEntry key=new DatabaseEntry();
						DatabaseEntry data=new DatabaseEntry();
						IntegerBinding.intToEntry(pair_id, key);
						TupleOutput out=new TupleOutput();
						if(indi1.id<indi2.id)
							{
							out.writeInt(indi1.id);
							out.writeInt(indi2.id);
							}
						else
							{
							out.writeInt(indi2.id);
							out.writeInt(indi1.id);
							}
						data.setData(out.getBufferBytes(),0,out.getBufferLength());
						if(this.pairIdIndividualId.put(txn, key, data)!=OperationStatus.SUCCESS)
							{
							throw new RuntimeException("Cannot insert pair");
							}
						this.count_pairs++;
						}
					}
				DatabaseEntry key=new DatabaseEntry();
				DatabaseEntry data=new DatabaseEntry();
				StringBinding.stringToEntry(tokens[5], key);
				
				LinkageRow lkrow=null;
				boolean rs_is_new;
				if(cursor.getSearchKey(key, data, LockMode.DEFAULT)==OperationStatus.SUCCESS)
					{
					lkrow=LinkageRow.BINDING.entryToObject(data);
					rs_is_new=false;
					}
				else
					{
					lkrow=new LinkageRow(this.count_pairs);
					rs_is_new=true;
					}
				
				
				lkrow.set(pair_id, new IBDStatus(
						(float)Double.parseDouble(tokens[6]),
						(float)Double.parseDouble(tokens[7]),
						(float)Double.parseDouble(tokens[8])
						));
				LinkageRow.BINDING.objectToEntry(lkrow, data);
				
				
				OperationStatus ret;
				if(rs_is_new)
					{
					ret=cursor.put(key, data);
					}
				else
					{
					ret=cursor.putCurrent(data);
					}
				
				if(ret!=OperationStatus.SUCCESS)
					{
					throw new RuntimeException("Cannot insert "+line);
					}
				++nInserted;
				}
			IOUtils.safeClose(in);
			
			LOG.info("Duration:"+(int)((System.currentTimeMillis()-start)/1000)/60.0+
					" mins. N="+nInserted+
					" N/sec="+(nInserted/((System.currentTimeMillis()-start)/1000.0))
					);
			}
		finally
			{
			if(cursor!=null) cursor.close();
			}
		
		}
	
	public void readDataPath(Transaction txn,File data) throws IOException
		{
		this.count_pairs=0;
		BufferedReader in=new BufferedReader(new FileReader(data));
		String line;
		int dataidx=0;
		while((line=in.readLine())!=null)
			{
			if(line.length()==0 || line.startsWith("#")) continue;
			setMeta(txn,"data["+(dataidx++)+"]",line);
			readDataFile(txn, new File(line));
			if(this.testing_flag) break;
			}
		IOUtils.safeClose(in);
		LOG.info("clean & sync");
		this.environment.sync();
		this.environment.compress();
		this.environment.cleanLog();
		}
	
	
	static private List<String> mappingSplit(String s)
		{
		List<String> tokens=new ArrayList<String>();
		char delim='\0';
		int i=0;
		while(i< s.length())
			{
			if(!(s.charAt(i)=='\'' || s.charAt(i)=='\"')) throw new RuntimeException("expected a quote or apos in "+s);
			delim=s.charAt(i);
			i++;
			if(i==s.length())
				{
				throw new RuntimeException("unfinished quoted string " + s);
				}
			int beg=i;
			int end=i;
			for(;;)
			    {
			    if(i==s.length())
					{
			    	throw new RuntimeException("unfinished quoted string " + s);
					}
			    if(s.charAt(i)==delim)
			     	{
					end=i;
					i++;
					break;
			     	}
				i++;
			    }
			String tok=s.substring(beg,end);
			//LOG("tokens["<< tokens.size()<< "]=" << tok );
			tokens.add(tok);
			if(i==s.length()) break;
			if(s.charAt(i)!=',') throw new RuntimeException("comma missing in "+ s);
			i++;
			if(!(s.charAt(i)=='\'' || s.charAt(i)=='\"')) throw new RuntimeException("expected a quote at the end of "+ s);
			}
		return tokens;
		}
	
	
	private static final String COL_PROBSET_ID ="Probe Set ID";
	private static final String COL_DBSNP_RS_ID ="dbSNP RS ID";
	private static final String COL_CHROMOSOME ="Chromosome";
	private static final String COL_POSITION ="Physical Position";
	private static final String COL_ALLELE_A ="Allele A";
	private static final String COL_ALLELE_B ="Allele B";
	private static final String COL_GENETIC_MAP ="Genetic Map";
	
	/** read an Axiom file mapping each marker */
	public void readMapping(Transaction txn,File f)throws IOException
		{
		count_markers=0;
		BufferedReader in=IOUtils.open(f);
		String line;
		
		setMeta(txn,"mapping.file",f.getAbsolutePath());
		int colIndexRsId=-1;
		int colChrom=-1;
		int colPosition=-1;
		int colProbSetId=-1;
		int colAlleleA=-1;
		int colAlleleB=-1;
		int colGeneticMap=-1;
		List<String> header=null;

		while((line=in.readLine())!=null)
			{
			if(line.length()==0 || line.charAt(0)=='#') continue;
			
			header=mappingSplit(line);

			for(int i=0;i< header.size();++i)
				{
				if(header.get(i).equals(COL_CHROMOSOME))
					{
					if(colChrom!=-1) throw new RuntimeException("Duplicate "+COL_CHROMOSOME);
					colChrom=i;
					}
				else if(header.get(i).equals(COL_POSITION))
					{
					if(colPosition!=-1) throw new RuntimeException("Duplicate "+COL_POSITION);
					colPosition=i;
					}
				else if(header.get(i).equals(COL_PROBSET_ID))
					{
					if(colProbSetId!=-1) throw new RuntimeException("Duplicate "+COL_PROBSET_ID);
					colProbSetId=i;
					}
				else if(header.get(i).equals(COL_DBSNP_RS_ID))
					{
					if(colIndexRsId!=-1) throw new RuntimeException("Duplicate "+COL_DBSNP_RS_ID);
					colIndexRsId=i;
					}
				else if(header.get(i).equals(COL_ALLELE_A))
					{
					if(colAlleleA!=-1) throw new RuntimeException("Duplicate "+COL_ALLELE_A);
					colAlleleA=i;
					}
				else if(header.get(i).equals(COL_ALLELE_B))
					{
					if(colAlleleB!=-1) throw new RuntimeException("Duplicate "+COL_ALLELE_B);
					colAlleleB=i;
					}
				else if(header.get(i).equals(COL_GENETIC_MAP))
					{
					if(colGeneticMap!=-1) throw new RuntimeException("Duplicate "+COL_GENETIC_MAP);
					colGeneticMap=i;
					}
				}
			if(colChrom==-1) throw new RuntimeException(COL_CHROMOSOME+" not found in "+line);
			if(colPosition==-1) throw new RuntimeException(COL_POSITION+" not found in "+line);
			if(colIndexRsId==-1) throw new RuntimeException(COL_DBSNP_RS_ID+" not found in "+line);
			if(colProbSetId==-1) throw new RuntimeException(COL_PROBSET_ID+" not found in "+line);
			if(colAlleleA==-1) throw new RuntimeException(COL_ALLELE_A+" not found in "+line);
			if(colAlleleB==-1) throw new RuntimeException(COL_ALLELE_B+" not found in "+line);
			if(colGeneticMap==-1) throw new RuntimeException(COL_GENETIC_MAP+" not found in "+line);
			break;
			}
		if(header==null || header.size()==0)
			{
			throw new RuntimeException("header not found in "+f);
			}
		
		while((line=in.readLine())!=null)
			{
			List<String> tokens=mappingSplit(line);
			if(tokens.size()!=header.size())
				{
				throw new RuntimeException("expected "+ header.size()
					+ " columns but got "
					+ tokens.size() +" in "+line);
				}
			if(tokens.get(colChrom).equals("---") ||
			   tokens.get(colPosition).equals("---"))
				  {
				  //ignore
				  continue;
				  }

			Marker marker=new Marker();

		
			marker.index=((count_markers)*-1)-1;//negate to avoid conflict at the end
			marker.name=tokens.get(colIndexRsId);
			if(getIBDByMarker(txn, marker.getName())==null) continue;

			marker.position=new ChromPosition(
					tokens.get(colChrom),
					Integer.parseInt(tokens.get(colPosition))
					);

			//genetic map
			int tripleslash=0;
			while(tripleslash<tokens.get(colGeneticMap).length())
				{
				/* example: token to parser:
				26.2497537844838 // D8S1827 // D8S549 // --- // --- /// 31.1074557729995 // D8S1827 // D8S549 // AFM107YA1 // AFM303ZC1 /// 28.7621103124195 // --- // --- // 259984 // 537009
				*/

				int doubeslash=tokens.get(colGeneticMap).indexOf(" // ",tripleslash);
				if(doubeslash==-1)
					{
					break;
					}
				marker.genMap.add(Float.parseFloat(tokens.get(colGeneticMap).substring(tripleslash,doubeslash)));
				tripleslash=doubeslash;


				tripleslash = tokens.get(colGeneticMap).indexOf(" /// ",tripleslash);
				if(tripleslash==-1)
					{
					break;
					}
				tripleslash+=5;
				}
			
			DatabaseEntry key=new DatabaseEntry();
			ChromPosition.BINDING.objectToEntry(marker.position,key);
			DatabaseEntry value=new DatabaseEntry();
			
			
			if(this.pos2marker.get(txn, key, value, LockMode.DEFAULT)==OperationStatus.SUCCESS)
				{
				throw new RuntimeException("Duplicate position for :\n"+marker+"\nand\n"+Marker.BINDING.entryToObject(value));
				}
			
			
			Marker.BINDING.objectToEntry(marker, value);
			if(this.pos2marker.put(txn, key, value)!=OperationStatus.SUCCESS)
				{
				throw new RuntimeException("Cannot insert "+line);
				}
			if(testing_flag && count_markers%1000==0)
				{
				LOG.info(marker+" \n\""+tokens.get(colGeneticMap)+"\"");
				}
			count_markers++;
			}
		in.close();
		
		finalizemarkerIndexes(txn);
		
		
		}
	
	public void readBed(Transaction txn,File f) throws IOException
		{
		count_markers=0;
		BufferedReader in=IOUtils.open(f);
		Pattern tab=Pattern.compile("[\t]");
		String line;
		
		setMeta(txn,"bed.file",f.getAbsolutePath());

		while((line=in.readLine())!=null)
			{
			if(line.length()==0 || line.charAt(0)=='#') continue;
			
			String tokens[]=tab.split(line);
			if(tokens.length<4)
				{
				System.err.println("Illegal number of columns in "+f+" :\n"+line);
				continue;
				}

			if(tokens[0].trim().length()==0)
				{
				System.err.println("Illegal chromosome in "+f+" :\n"+line);
				continue;
				}
			
			if(tokens[1].trim().length()==0)
				{
				System.err.println("Illegal position in "+f+" :\n"+line);
				continue;
				}
			
			if(tokens[3].trim().length()==0)
				{
				System.err.println("Illegal rs## name in "+f+" :\n"+line);
				continue;
				}

			Marker marker=new Marker();

		
			marker.index=((count_markers)*-1)-1;//negate to avoid conflict at the end
			marker.name=tokens[3].trim();
			
			
			if(getIBDByMarker(txn, marker.getName())==null)
				{
				//LOG.info("marker "+marker+" not used");
				continue;
				}
			
			marker.position=new ChromPosition(
					tokens[0].trim(),
					Integer.parseInt(tokens[1].trim())
					);
			
			DatabaseEntry key=new DatabaseEntry();
			ChromPosition.BINDING.objectToEntry(marker.position,key);
			DatabaseEntry value=new DatabaseEntry();
			
			if(this.pos2marker.get(txn, key, value, LockMode.DEFAULT)==OperationStatus.SUCCESS)
				{
				throw new RuntimeException("Duplicate position for :\n"+marker+"\nand\n"+Marker.BINDING.entryToObject(value));
				}
			
			Marker.BINDING.objectToEntry(marker, value);
			if(this.pos2marker.put(txn, key, value)!=OperationStatus.SUCCESS)
				{
				throw new RuntimeException("Cannot insert "+line);
				}
			
			
			if(testing_flag && count_markers%1000==0)
				{
				LOG.info(line);
				}
			count_markers++;
			}
		in.close();
		finalizemarkerIndexes(txn);
		}
	
	/** called by readBed or ReadMapping, create an index of markers */
	private void finalizemarkerIndexes(Transaction txn)
		{
		Cursor c=null;
		try
			{
			count_markers=0;
			DatabaseEntry key=new DatabaseEntry();
			DatabaseEntry data=new DatabaseEntry();
			
			c=pos2marker.openCursor(txn, null);
			while(c.getNext(key, data, LockMode.DEFAULT)==OperationStatus.SUCCESS)
				{
				Marker m=Marker.BINDING.entryToObject(data);

				m.index=count_markers;
				Marker.BINDING.objectToEntry(m, data);
				if(c.putCurrent(data)!=OperationStatus.SUCCESS)
					{
					throw new RuntimeException("boum");
					}
				++count_markers;
				}
			}
		finally
			{
			if(c!=null) c.close();
			}
		
		//check all markers in linkage have a position
		c=null;

		try
			{
			DatabaseEntry key=new DatabaseEntry();
			DatabaseEntry data1=new DatabaseEntry();
			
			DatabaseEntry data2=new DatabaseEntry();
			c=this.marker2linkage.openCursor(txn, null);
	
			while(c.getNext(key, data1, LockMode.DEFAULT)==OperationStatus.SUCCESS)
				{
				if(this.name2marker.get(txn,key, data2,LockMode.DEFAULT)!=OperationStatus.SUCCESS)
					{
					throw new RuntimeException("marker:\""+StringBinding.entryToString(key)+"\" used but not mapped???!!!");
					}
				}
			}
		finally
			{
			if(c!=null) c.close();
		
			}
		}

	public void readReskin(Transaction txn,File f) throws IOException
		{
		setMeta(txn,"reskin.file",f.getAbsolutePath());
		BufferedReader in=IOUtils.open(f);
		String line;
		Pattern tab=Pattern.compile("[\t ]");
		int nFound=0;
		while((line=in.readLine())!=null)
				{
				if(line.length()==0 || line.charAt(0)=='#') continue;
				String tokens[]=tab.split(line);
				if(tokens.length!=12) throw new RuntimeException("Expected 12 columns in "+ line);
				Individual indi1= getIndividualbyName(txn, tokens[0],tokens[1]);
				if(indi1==null)
					{
					LOG.info("Cannot find individual1 "+tokens[0]+"/"+tokens[1]);
					continue;
					}
				Individual indi2= getIndividualbyName(txn, tokens[0],tokens[2]);
				if(indi2==null)
					{
					LOG.info("Cannot find individual2 "+tokens[0]+"/"+tokens[2]);
					continue;
					}
				
				int pair_id=findPairIdByIndividuals(txn, indi1, indi2);
				if(pair_id==-1)
					{
					LOG.info("Cannot find pair id for "+indi1+"/"+indi2);
					continue;
					}
				
				double ibd2=Double.parseDouble(tokens[11-2]);
				double ibd1=Double.parseDouble(tokens[11-1]);
				double ibd0=Double.parseDouble(tokens[11-0]);
				IBDStatus status=new IBDStatus((float)ibd0, (float)ibd1, (float)ibd2);
				
				DatabaseEntry key=new DatabaseEntry();
				IntegerBinding.intToEntry(pair_id,key);
				DatabaseEntry value=new DatabaseEntry();
				IBDStatus.BINDING.objectToEntry(status, value);
				if(this.pairid2theo.put(txn, key, value)!=OperationStatus.SUCCESS)
					{
					throw new RuntimeException("cannot insert "+line);
					}
				++nFound;
				}
		IOUtils.safeClose(in);
		if(nFound==0) throw new RuntimeException("no theo ibd status found in "+f);
		}
	
	
	/*
	static public void main(String args[]) throws Exception
		{
		Transaction txn=null;
		Model.LOG.setLevel(Level.ALL);
		Model m=new Model();
		m.testing_flag=false;
		File home=new File("/BELENOS/data/project/20111031jibdstatus/_ignore.backup/data");
		m.open(home,false);
		
		m.readPedigree(txn, new File("/BELENOS/data/project/20110922ibdbrugada/data/_ignore.backup/brugada7.fam"));
		m.readReskin(txn, new File("/BELENOS/data/project/20110922ibdbrugada/data/_ignore.backup/ibdldreskin_brugada7"));
		m.readDataPath(txn, new File("/home/lindenb/src/bdbidb/data/ibdpath.txt"));
		m.readMapping(txn, new File("/home/lindenb/src/bdbidb/data/Axiom_GW_Hu_SNP.r2.na30.annot.csv.gz"));
		m.close();
		m.open(home,true);
		
		LOG.info(""+m.count_markers);
		LOG.info(""+m.getMarkerByIndex(null, m.count_markers-1));
		m.close();
		}*/
	
	}
